package com.example.mq.mqserver.datacenter;

import com.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*使用这个类来统一管理内存中的所有数据
* 该类后续提供的一些方法，可能会在多线程环境下被使用，因此要注意线程安全问题
* */
public class MemoryDataCenter {
    //key 是 exchangeName， value 是 exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key 是 queueName，value 是MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //String 是 exchangeName，嵌套中的ConcurrentHashMap 中的 String 是 queueName，Binding 是 绑定对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId，value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是队列的名字，value 是一个 Message 的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    //针对交换机在内存中的操作的封装
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功！exchangeName= " + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！ exchangeName= " + exchangeName);
    }

    // 针对队列在内存中的操作的封装
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功！ queueName= " + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！ queueName= " + queueName);
    }

    //针对绑定 在内存中的操作 的封装
    public void insertBinding(Binding binding) throws MqException {

 //        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }
        //这段逻辑就等价于上述代码的逻辑，只是用一个方法就搞定了
        // 先使用 exchange 查一下，对应的哈希表是否存在，不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        //此处单拿出一个操作，是线程安全的，但是现在是 get 和 put 是放在一起的，此时就可能第一个线程判断是空的，紧接着还没
        //来得及 put，此时第二个线程既 get 又 put，然后再切换到第一个线程再进行put，此时就不符合要求了。
        //所以针对这两步操作就需要加锁让两步操作变成原子性的操作
        //加锁的对象就拿 bindingMap 作为锁对象即可，如果拿bindingsMap 来加锁，此时加锁的粒度就太大了，当前操作是
        //针对 bindingMap 进行查询和插入，所以针对 bindingMap 加锁即可
        synchronized (bindingMap){
            // 再根据 queueName 查一下，如果已经存在就抛出异常，不存在才能进行创建
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在！ exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新的绑定添加成功！exchangeName="
                + binding.getExchangeName() + ", queueName= " + binding.getQueueName());
    }

    //获取绑定写两个版本：
    //1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    //2. 根据 exchangeName 获取到所有的 Binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {// 如果 bindingMap 为空就说明此时没有绑定任何的队列
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            //说明该交换机没有和任何队列进行绑定， 报错
            throw new MqException("[MemoryDataCenter] 绑定不存在！exchangeName=" + binding.getExchangeName()
            + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功！exchangeName="
                + binding.getExchangeName() + ", queueName= " + binding.getQueueName());
    }

    //添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功！messageId= " + message.getMessageId());
    }

    //根据id 查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    //根据 id 删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除！ messageId= " + messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        // 把消息放到对应的队列数据结构中，
        // 先根据队列的名字，找到该队列对应的链表
//       LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//       if (messages == null) {
//           messages = new LinkedList<>();
//           queueMessageMap.put(queue.getName(), messages);
//       }
        //上述的代码还是一样可以使用 computeIfAbsent 方法来代替上述逻辑，这个方法的意思就是先根据队列的名字查询这个链表
        //如果没有查询到这个链表就创建一个链表然后加入到队列中，而且这个方法本身也是线程安全的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k-> new LinkedList<>());
        // 再把消息数据加到 messages 里面
        // 注：LinkedList 不是线程安全的 所以在添加消息的时候需要进行加锁
        synchronized (messages) {
            messages.add(message);
            //在这里把该消息也往消息中心插入一下, 假设如果 message 已经在消息中心存在，重复插入内容也没关系
            //主要就是相同 messageId，对应的 message 的内容一定是一样的 （服务器代码不会对 Message 内容做修改：basicProperties 和 body）
            addMessage(message);
            System.out.println("[MemoryDataCenter] 消息被投递到队列中！messageId=" + message.getMessageId());
        }

    }
    //从队列中取消息
    public Message pollMessage(String queueName) {
        //根据队列名，查找一下对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        // 如果没找到，说明队列中没有任何消息
        if (messages.size() == 0) {
            return null;
        }
        synchronized (messages) {  //保证获取链表长度和删除头元素都是线程安全的
            //链表中有元素，就进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出！ messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取指定队列中的消息个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0; // 队列中没有消息
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        //同样的也是先根据队列的名字找到对应的未确认消息的哈希表，如果找不到就创建一个
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        //然后把未确认的消息对应的哈希表添加到整体的 未确认消息中心（整个的ConcurrentHashMap）中
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列！messageId=" + message.getMessageId());
        //上述逻辑关联性不强，就不用加锁了   (为啥不加锁？？)
    }

    // 删除未确认的消息（消息已经确认了）
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) return;
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认消息队列中删除！message=" + messageId);
    }

    // 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) return null;
        return messageHashMap.get(messageId);
    }

    // 这个方法就是从硬盘上读取数据，把硬盘中之前持久化存储的各个维度的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0. 先清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        queueMessageMap.clear();
        //1. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();

        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        //2. 恢复虽有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        //3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k-> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        //4. 恢复所有的消息数据
        // 遍历所有的队列，根据每个队列的名字，获取到所有的消息
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
    }
    // 注：针对”未确认的消息“ 这部分内存中的数据，不需要从硬盘恢复，之前考虑硬盘存储的时候，也没设定这一块，
    // 一旦在等待 ack 的过程中，服务器重启了，此时这些 ”未被确认的消息“，就恢复成”未被取走的消息“，
    // 这个消息在硬盘上存储的时候，就是当作 ”未被取走“

}
